package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

import scala.reflect.ClassTag

/**
  * Created by Administrator on 2017/5/2 0002.
  * 在执行subgraph()进行边过滤后，下面的代码提供了删除所有单独（孤立）顶点的实现方法。
  */
object TestRemoveSingletons {
  def removeSingletons[VD:ClassTag, ED:ClassTag](g:Graph[VD,ED]) =
    Graph(g.triplets.map(et =>(et.srcId, et.srcAttr))
          .union(g.triplets.map(et => (et.dstId, et.dstAttr)))
          .distinct, g.edges)


  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val vertices = sc.makeRDD(Seq((1L, "Ann"), (2L, "Bill"), (3L, "Charles"), (4L, "Dianne")))
    val edges = sc.makeRDD(Seq(Edge(1L, 2L, "is-friends-with"), Edge(1L,3L, "is-friends-with"), Edge(4L, 1L, "has-blocked"),
                                Edge(2L, 3L, "has-blocked"), Edge(3L, 4L, "has-blocked")))
    val originalGraph = Graph(vertices, edges)
    val subgraph = originalGraph.subgraph( et => et.attr == "is-friends-with")

    //展示子图中的顶点，包括Dianne
    subgraph.vertices.foreach(println)

    //调用removeSingletons函数，并展示结果对应的顶点
    removeSingletons(subgraph).vertices.foreach(println)
  }
}
